package com.example.flinkcourse.lesson4.source;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.connector.jdbc.JdbcSource;
import org.apache.flink.streaming.api.functions.source.Source;
import org.apache.flink.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * JDBC 数据源工厂类
 * 用于创建 JDBC 数据源，支持数据库查询
 */
public class JdbcSourceFactory {
    private static final Logger LOG = LoggerFactory.getLogger(JdbcSourceFactory.class);

    /**
     * 创建 JDBC 数据源
     * @param url 数据库连接 URL
     * @param query SQL 查询语句
     * @return JDBC 数据源
     */
    public static Source<Row> createJdbcSource(String url, String query) {
        LOG.info("Creating JDBC source for URL: {}", url);
        
        JdbcSource<Row> source = JdbcSource.<Row>builder()
            .setDBUrl(url)
            .setQuery(query)
            .setRowTypeInfo(Types.ROW(
                Types.STRING,  // id
                Types.STRING,  // type
                Types.LONG,    // timestamp
                Types.DOUBLE   // amount
            ))
            .build();
        
        LOG.info("JDBC source created successfully");
        return source;
    }
    
    /**
     * 创建带认证的 JDBC 数据源
     * @param url 数据库连接 URL
     * @param username 用户名
     * @param password 密码
     * @param query SQL 查询语句
     * @return JDBC 数据源
     */
    public static Source<Row> createJdbcSourceWithAuth(String url, String username, String password, String query) {
        LOG.info("Creating JDBC source with authentication for URL: {}", url);
        
        JdbcSource<Row> source = JdbcSource.<Row>builder()
            .setDBUrl(url)
            .setUsername(username)
            .setPassword(password)
            .setQuery(query)
            .setRowTypeInfo(Types.ROW(
                Types.STRING,  // id
                Types.STRING,  // type
                Types.LONG,    // timestamp
                Types.DOUBLE   // amount
            ))
            .build();
        
        LOG.info("JDBC source with authentication created successfully");
        return source;
    }
} 